package com.wenge.datagroup.storage.util

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.wenge.datagroup.storage.dataprocess._
import com.wenge.datagroup.storage.future.AsyncProcess
import org.apache.commons.codec.digest.DigestUtils
import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.util.Time.now
import org.apache.log4j.Logger

import java.lang.Exception
import javax.annotation.Nonnull
import java.net.URL
import java.sql.ResultSet
import java.text.SimpleDateFormat
import java.util
import java.util.{Date, Objects, Random, HashSet => JavaHashSet, List => JavaList, Map => JavaMap}
import java.util.concurrent.ConcurrentHashMap
import scala.collection.mutable.ArrayBuffer
import scala.language.postfixOps
import scala.util.control.Exception

/**
 * @author
 * Depp
 */
class TransformUtil(
                     druidDataSource: DruidDataSource,
                     multilingualLanguage: MultilingualLanguage,
                     multilingualTranslation: MultilingualTranslation) {
  lazy val EMPTY_JSONARRAY = new JSONArray()
  lazy val EMPTY_ARRAYLIST = new util.ArrayList[String]
  lazy val random = new Random
  private val logger: Logger = Logger.getLogger(classOf[TransformUtil])


  /**
   * md5加密
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   */
  def md5Hex(orgRecord: JSONObject, orgField: String): String = {
    val orgValue: String = orgRecord.getString(orgField)
    if (StringUtils.isNotBlank(orgValue)) {
      return DigestUtils.md5Hex(orgValue)
    }
    DigestUtils.md5Hex(String.valueOf(random.nextInt()))
  }

  /**
   * _id 拼接值
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   * @param preValue
   * 前缀
   * @return
   */
  def concat(orgRecord: JSONObject,
             orgField: String, preValue: String): String = {
    //val _keyValue: String = md5Hex(orgRecord, orgField)
    /* val _keyValue: String = map.get("_key").asInstanceOf[String]
     if (StringUtils.isNotBlank(_keyValue)) {
       // 通用新增境内外渠道
       if ("tygksy_".equals(preValue)) {
         return preValue + _keyValue + orgRecord.getString("gregion")
       } else {
         return preValue + _keyValue
       }
     }
     StringUtils.EMPTY*/
    return preValue
  }

  // 移动应用商店渠道 key拼接 appstore_
  def concatApp(
                 orgRecord: JSONObject,
                 map: ConcurrentHashMap[String, Any],
                 preValue: String): String = {

    val ptValue: String = orgRecord.getString("pt")
    val PtStr: String = ptValue.split(" ")(0).replaceAll("-", "")
    val sdf = new SimpleDateFormat("yyyyMMdd")

    if (preValue.equals("appstore_")) {
      val appIdValue: String = orgRecord.getString("appid")
      val storeValue: String = orgRecord.getString("store")
      val regionValue: String = orgRecord.getString("region")

      val s = preValue + appIdValue + "_" + storeValue + "_" + regionValue
      val result = s.replaceAll("null", "")
      return result
    }
    if ("dvid".equals(preValue)) {
      val appIdValue: String = orgRecord.getString("appid")
      val storeValue: String = orgRecord.getString("store")
      val regionValue: String = orgRecord.getString("region")

      val s = "appstore_" + appIdValue + "_" + storeValue + "_" + regionValue
      val result = s.replaceAll("null", "")
      return result + "_" + PtStr
    }

    if (preValue.equals("apprank_")) {
      val url: String = orgRecord.getString("source_url")
      val keyValue: String = DigestUtils.md5Hex(url)

      if (StringUtils.isNotBlank(ptValue.toString)) {

        return preValue + keyValue + "_" + PtStr
      } else {
        return preValue + keyValue + "_" + sdf.format(new Date(now()))
      }

    }

    ""
  }

  /**
   * 获取语种
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   * @return
   */
  def getLang(orgRecord: JSONObject, @Nonnull orgField: String): String = { // 有原始字段 无值 有值
    var langValue: String = null
    if (StringUtils.isNotBlank(orgField)) {
      var langKey: Any = null
      if (orgField.split(",").length > 1) {
        langKey = concatValues(orgRecord, orgField)._1
      } else langKey = orgRecord.get(orgField)
      if (langKey != null) {
        langKey match {
          case _: Int => langValue = LangMappingUtil.lanMapping.get(langKey.asInstanceOf[Int])
          case _: String =>
            langValue = multilingualLanguage.analysis(langKey.asInstanceOf[String], "2221")
          case _ =>
        }
      }
    }
    // 无原始字段
    if (StringUtils.isBlank(langValue)) {
      StringUtils.EMPTY
    } else {
      langValue
    }
  }

  /**
   * 关键词抽取
   *
   * @param map
   * 新记录
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 关键词提取对应的字段值(旧记录) eg: title,content
   * @return
   */
  def keywordsExtract(
                       orgRecord: JSONObject,
                       map: ConcurrentHashMap[String, Any],
                       orgField: String): JSONArray = {
    val langKey: String = map.get("lang").asInstanceOf[String]
    var lang = LangMappingUtil.aiLanMapping.get(langKey)
    val word: String = concatValues(orgRecord, orgField)._1
    val orgFields: Array[String] = concatValues(orgRecord, orgField)._2
    if (orgFields == null || orgFields.isEmpty) {
      return EMPTY_JSONARRAY
    }
    if (StringUtils.isBlank(lang)) {
      lang = multilingualLanguage.analysis(word, "121")
    }
    val keywords: JSONArray =
      MultilingualKeywords.analysis(
        orgRecord.getString(orgFields(0)),
        orgRecord.getString(orgFields(1)),
        lang)
    if (keywords == null) EMPTY_JSONARRAY else keywords
  }

  /**
   * 实体抽取
   *
   * @param orgRecord
   * 旧记录
   * @param map
   * 新纪录
   * @param orgField
   * 实体抽取的内容对应的旧字段
   * @param finalField
   * 新字段
   * @return
   */
  def entityExtract(
                     orgRecord: JSONObject,
                     map: ConcurrentHashMap[String, Any],
                     orgField: String,
                     finalField: String): ConcurrentHashMap[String, Any] = {
    map.put(finalField, EMPTY_JSONARRAY)
    val word: String = concatValues(orgRecord, orgField)._1
    if (word.isEmpty) {
      return map
    }
    val entityAnalysisResult: JavaMap[_, _] =
      MultilingualEntity.analysis(word)
    if (entityAnalysisResult != null) {
      var entityType: Any = null
      finalField match {
        case "vpers" =>
          entityType = entityAnalysisResult.get("PER")
        case "vorg" =>
          entityType = entityAnalysisResult.get("ORG")
        case "vrgn" =>
          entityType = entityAnalysisResult.get("LOC")
        case _ =>
      }
      val vector: util.Set[String] = new JavaHashSet[String]
      if (!Objects.isNull(entityType)) {
        val per: JSONArray = JSON.parseArray(entityType.toString)
        import scala.collection.JavaConversions._
        for (o <- per) {
          vector.add(o.toString)
        }
        map.put(finalField, vector)
      }
    }
    map
  }

  /**
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   * @return
   */
  private def concatValues(
                            orgRecord: JSONObject,
                            orgField: String): (String, Array[String], Array[String]) = {
    var word: String = StringUtils.EMPTY
    if (StringUtils.isBlank(orgField)) {
      return (word, null, null)
    }
    val orgFields: Array[String] = orgField.split(",", -1)
    val contArr: ArrayBuffer[String] = new ArrayBuffer[String]()
    var i = 0
    for (field <- orgFields) {
      val str: String = orgRecord.getString(field.trim)
      if (StringUtils.isNotBlank(str)) {
        word += str
      }
      contArr += str
      i += 1
    }
    (word, orgFields, contArr.toArray)
  }

  /**
   * 情感倾向分析
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 情感识别内容对应的旧字段
   */
  def emoExtract(orgRecord: JSONObject, orgField: String): Long = {
    var analysisEmotionResult: String =
      MultilingualEmotion.analysis(concatValues(orgRecord, orgField)._1)
    if (analysisEmotionResult == null) { // 默认给到中立
      analysisEmotionResult = "0.5"
    }
    Math.round(analysisEmotionResult.toDouble * 100.round)
  }

  /**
   * 翻译
   *
   * @param orgRecord
   * 旧记录
   * @param map
   * 新纪录
   * @param orgField
   * 旧字段
   * @param finalField
   * 新字段
   * @return
   */
  def translationProcess(
                          orgRecord: JSONObject,
                          map: ConcurrentHashMap[String, Any],
                          orgField: String,
                          finalField: String): ConcurrentHashMap[String, Any] = {
    if (map.containsKey(finalField)) {
      if ("" != map.get(finalField)) {
        logger.info("start Map contains :" + finalField + " : " + map.get(finalField))
        map.put(finalField, "")
        logger.info("end Map contains :" + finalField + " : " + map.get(finalField))
      }
    }

    var filedLang: String = map.get("lang").asInstanceOf[String]
    val str: String = orgRecord.getString(orgField)

    if (StringUtils.isBlank(str)) {
      map.put(finalField, StringUtils.EMPTY)
      return map
    }

    //var filedLang: String = ""
    if (StringUtils.isBlank(filedLang)) { // 语种识别
      try {
        filedLang = multilingualLanguage.analysis(orgRecord.getString(orgField), "1111")
      } catch {
        case e: Exception => println(e.printStackTrace())
      } finally {
        if (StringUtils.isBlank(filedLang)) {
          map.put(finalField, str)
        }
      }
    }

    if (filedLang == "zh") { // 中文不做文章直译 给默认值
      map.put(finalField, str)
    } else if (StringUtils.isBlank(filedLang)) {
      map.put(finalField, str)
    } else {
      if (str != null) {
        var contentTraResult: String = ""
        try {
          contentTraResult = multilingualTranslation.analysis(str, filedLang)
        } catch {
          case e: Exception => println(e.printStackTrace())
        } finally {

          if (StringUtils.isNotBlank(contentTraResult)) {
            map.put(finalField, contentTraResult)
          } else {
            val afterFilter: String = SpaceFilter.filter(EmojiFilter.filter(HtmlFilter.filter(str)))
            var secondResult: String = ""
            try {
              secondResult = multilingualTranslation.analysis(afterFilter, filedLang)
            } catch {
              case e: Exception => println(e.printStackTrace())
            } finally {
              if (StringUtils.isNotBlank(secondResult)) {
                map.put(finalField, secondResult)
              } else {
                map.put(finalField, str)
              }
            }
          }
        }
      }
    }

    val value: String = map.get(finalField).asInstanceOf[String]
    if (StringUtils.isBlank(value)) {
      map.put(finalField, str)
    }
    map
  }

  def wikiTranslationProcess(
                              orgRecord: JSONObject,
                              map: ConcurrentHashMap[String, Any],
                              orgField: String,
                              finalField: String): ConcurrentHashMap[String, Any] = {
    var filedLang: String = map.get("lang").asInstanceOf[String]
    val str: String = orgRecord.getString(orgField)

    if (StringUtils.isBlank(str)) {
      map.put(finalField, StringUtils.EMPTY)
      return map
    }
    if (filedLang == "zh") { // 中文不做文章直译 给默认值
      map.put(finalField, str)
    } else if (StringUtils.isBlank(filedLang)) {
      map.put(finalField, str)
    } else {
      if (str != null) {
        var contentTraResult: String = ""
        try {
          contentTraResult = multilingualTranslation.analysis(str, filedLang)
        } catch {
          case e: Exception => println(e.printStackTrace())
        } finally {

          if (StringUtils.isNotBlank(contentTraResult)) {
            map.put(finalField, contentTraResult)
          } else {
            val afterFilter: String = WikiFilter.filter(SpaceFilter.filter(EmojiFilter.filter(HtmlFilter.filter(str))))
            var secondResult: String = ""
            try {
              secondResult = multilingualTranslation.analysis(afterFilter, filedLang)
            } catch {
              case e: Exception => println(e.printStackTrace())
            } finally {
              if (StringUtils.isNotBlank(secondResult)) {
                map.put(finalField, secondResult)
              } else {
                map.put(finalField, str)
              }
            }
          }
        }
      }
    }

    val value: String = map.get(finalField).asInstanceOf[String]
    if (StringUtils.isBlank(value)) {
      map.put(finalField, str)
    }
    map
  }

  /**
   * 自动摘要 先调用摘要接口再对生成的摘要做翻译
   *
   * @param orgRecord
   * 旧记录
   * @param map
   * 新纪录
   * @param orgField
   * 旧字段 eg: 1、title, 2、,content 3、title,content
   * @param finalField
   * 新字段
   */
  def autoAbstractProcess(
                           orgRecord: JSONObject,
                           map: ConcurrentHashMap[String, Any],
                           orgField: String,
                           finalField: String): ConcurrentHashMap[String, Any] = {
    var abstractAnalysis: String = null
    val word: String = concatValues(orgRecord, orgField)._1
    val contArr: Array[String] = concatValues(orgRecord, orgField)._3
    if (word.isEmpty) {
      map.put(finalField, StringUtils.EMPTY)
      map.put(finalField + "zh", StringUtils.EMPTY)
      return map
    } else {
      if (StringUtils.isBlank(contArr(1))) {
        abstractAnalysis = MultilingualAbstract.analysis(contArr(0), StringUtils.EMPTY)
      } else if (StringUtils.isBlank(contArr(0))) {
        abstractAnalysis = MultilingualAbstract.analysis(StringUtils.EMPTY, contArr(1))
      } else abstractAnalysis = MultilingualAbstract.analysis(contArr(0), contArr(1))
    }
    map.put(finalField, abstractAnalysis)

    var filedLang: String = map.get("lang").asInstanceOf[String]
    if (StringUtils.isBlank(filedLang)) { // 语种识别
      filedLang = multilingualLanguage.analysis(orgRecord.getString(abstractAnalysis), "1111")
    }
    if (filedLang == "zh") { // 中文不做摘要直译 给默认值
      map.put(finalField + "zh", abstractAnalysis)
    } else {
      val contentTraResult: String =
        multilingualTranslation.analysis(abstractAnalysis, filedLang)
      if (StringUtils.isNotBlank(contentTraResult)) {
        map.put(finalField + "zh", contentTraResult)
      } else {
        map.put(finalField + "zh", StringUtils.EMPTY)
      }
    }

    map
  }

  /**
   * 分词
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   * @param finalField
   * 新字段
   */
  def splitWordProcess(
                        orgRecord: JSONObject,
                        orgField: String,
                        finalField: String): JavaList[String] = {
    if (orgRecord.getString(orgField) != null) {
      val titleResult: JavaList[String] =
        MultilingualSpiltWords.analysis(clearStr(orgRecord.getString(orgField)))
      titleResult
    } else EMPTY_ARRAYLIST
  }

  /**
   * 移动应用商店渠道字段转换 赋值需判断
   *
   * @param orgRecord
   * 旧记录
   * @param orgField
   * 旧字段
   * @return
   */
  def otherString(orgRecord: JSONObject, orgField: String): String = {
    val value: String = orgRecord.getString(orgField)
    if (StringUtils.isNotBlank(value) && "ios".equals(value)) {
      return "ios"
    }
    "android"
  }

  def otherBoolean(orgRecord: JSONObject, orgField: String): Boolean = {
    val value: String = orgRecord.getString(orgField)
    if ("free".equals(orgField)) {
      if (StringUtils.isNotBlank(value) && "免费".equals(value)) {
        return true
      }
      return false
    }

    if ("hot".equals(orgField)) {
      val hotValue: Boolean = orgRecord.getBoolean(orgField)
      val str: String = orgRecord.getString("hot")
      if (hotValue) {
        return hotValue
      }
      return false
    }
    false
  }

  /**
   * 清理字符串
   *
   * @param str
   * 目标字符串
   * @return
   * 清理后的字符串
   */
  def clearStr(str: String): String = {
    if (StringUtils.isEmpty(str)) return str
    val resultStr = str.replaceAll("\\\n|\\\t|\\ |\\　　|\\\r", StringUtils.EMPTY)
    resultStr.trim
  }

  /**
   * 清理字符串
   *
   * @param orgRecord
   * 一条记录
   * @param orgField
   * 原始字段名
   * @param finalField
   * 最终字段名
   * @return
   * 清理后的记录
   */
  def mappingMegType(orgRecord: JSONObject, orgField: String, finalField: String): Int = {
    val orgValue: String = orgRecord.getString(orgField)
    val finalValue: Int = orgValue match {
      case "0" => 1
      case "1" => 3
      case _ => -1
    }
    finalValue
  }

  /**
   * 提取域名
   *
   * @param record
   * 一条记录
   * @param orgField
   * 原始字段名
   * @param finalField
   * 最终字段名
   * @return
   * 清理后的记录
   */
  def extractHost(orgRecord: JSONObject, orgField: String, finalField: String): String = {
    val url: String = orgRecord.getString(orgField)
    val host: String = url match {
      case null => null
      case _ => new URL(url).getHost
    }
    host
  }


  /**
   * 获取getAlexaInfo
   *
   * @param orgRecord
   * 旧记录
   * @param map
   * 新纪录
   * @param hostName
   * 不同渠道域名对应字段名
   */
  def getAlexaInfo(
                    orgRecord: JSONObject,
                    map: ConcurrentHashMap[String, Any],
                    hostName: String): ConcurrentHashMap[String, Any] = {
    // 如果旧字段中host所属字段有值，取该值，否则取新的值
    val orgHostValue: String = orgRecord.getString(hostName)
    val hostValue: String = map.get(hostName).asInstanceOf[String]
    val subHost = if (StringUtils.isNotBlank(orgHostValue)) orgHostValue else hostValue
    if (StringUtils.isNotBlank(subHost)) {
      val sqlProxy = new SqlProxy()
      val client = druidDataSource.dataSource.getConnection
      try {
        sqlProxy.executeQuery(
          client,
          "select ranking from `centre_collect`.`ctms_alexadate` where hostName=?",
          Array(subHost),
          new QueryCallback {
            override def process(rs: ResultSet): Unit = {
              // 判断对应数据库中是否有值
              if (rs.isBeforeFirst) {
                while (rs.next()) {
                  val ranking = rs.getLong(1)
                  map.put("alexaInfo", ranking)
                  println("mysql获取ranking:" + ranking)
                }
                rs.close() // 关闭资源
              } else {
                println("mysql获取ranking为空")
                map.put("alexaInfo", 0)
              }
            }
          }
        )
      } catch {
        case e: Exception =>
          e.printStackTrace()
      } finally {
        sqlProxy.shutdown(client)
      }
    } else {
      println("mysql获取ranking未查询，直接赋值")
      map.put("alexaInfo", 0)
    }

    map
  }


  /**
   * 获取网站名称 新版
   *
   * @param orgRecord
   * 旧记录
   * @param map
   * 新纪录
   * @param hostName
   * 不同渠道域名对应字段名
   */
  def getIsnNew(
                 orgRecord: JSONObject,
                 map: ConcurrentHashMap[String, Any],
                 hostName: String): ConcurrentHashMap[String, Any] = {
    // 如果旧字段中host所属字段有值，取该值，否则取新的值
    val orgHostValue: String = orgRecord.getString(hostName)

    val hostValue: String = map.get(hostName).asInstanceOf[String]
    val subHost = if (StringUtils.isNotBlank(orgHostValue)) orgHostValue else hostValue

    if (StringUtils.isNotBlank(subHost)) {
      val sqlProxy = new SqlProxy()
      val client = druidDataSource.dataSource.getConnection
      try {
        sqlProxy.executeQuery(
          client,
          "select auto_id,name from `centre_collect`.`host_info` where host=?",
          Array(subHost),
          new QueryCallback {
            override def process(rs: ResultSet): Unit = {
              // 判断对应数据库中是否有值
              if (rs.isBeforeFirst) {
                while (rs.next()) {
                  val iSid = rs.getLong(1)
                  val iSn = rs.getString(2)
                  map.put("i_sid", iSid)
                  map.put("i_sn", iSn)
                }
                rs.close() // 关闭资源
              } else {
                // 往数据库填充
                val cnt = sqlProxy.executeUpdate(
                  client,
                  "insert into `centre_collect`.`host_info` (`host`,`name`) values(?,?)",
                  Array(subHost, subHost))
                // 查询并取生成的auto_id返回 优化
                sqlProxy.executeQuery(
                  client,
                  "select auto_id,name from `centre_collect`.`host_info` where host=?",
                  Array(subHost),
                  new QueryCallback {
                    override def process(rs: ResultSet): Unit = {
                      while (rs.next()) {
                        val iSid = rs.getLong(1)
                        val iSn = rs.getString(2)
                        map.put("i_sid", iSid)
                        map.put("i_sn", iSn)
                      }
                      rs.close()
                    }
                  }
                )
              }
            }
          }
        )
      } catch {
        case e: Exception =>
          e.printStackTrace()
      } finally {
        sqlProxy.shutdown(client)
      }
    } else {
      map.put("i_sid", 0)
      map.put("i_sn", StringUtils.EMPTY)
    }
    map
  }
}
